package com.fzf.org.mq.consumer;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fzf.org.dao.entities.AuditDO;
import com.fzf.org.dao.entities.BlogDO;
import com.fzf.org.dao.mapper.AuditMapper;
import com.fzf.org.dao.mapper.BlogMapper;
import com.fzf.org.enums.AuditEnums;
import com.fzf.org.mq.idempotent.MessageQueueIdempotentHandler;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * 博客审核消息消费者
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class BlogReviewerConsumer implements StreamListener<String, MapRecord<String, String, String>> {

    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;
    private final RestTemplate restTemplate;
    private final ObjectMapper objectMapper;
    private final BlogMapper blogMapper;
    private final AuditMapper auditMapper;

    private String reviewApiUrl = "http://localhost:8080/blog/audit/mock-review"; // TODO 博客审核API地址

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        // 处理空消息情况
        if (message == null) {
            log.error("Received null message");
            return;
        }

        // 处理空ID情况
        if (message.getId() == null) {
            log.error("Received message with null ID: {}", message);
            return;
        }

        String messageId = message.getId().getValue();
        Map<String, String> body = message.getValue();

        // 处理type=init的初始化消息
        if (body != null && "init".equals(body.get("type"))) {
            log.info("Received stream initialization message, ignoring: {}", messageId);
            return;
        }

        log.info("Received blog review message: {}, body: {}", messageId, body);

        try {
            // 1. 检查消息是否正在被消费(幂等处理)
            if (messageQueueIdempotentHandler.isMessageBeingConsumed(messageId)) {
                if (messageQueueIdempotentHandler.isAccomplish(messageId)) {
                    log.info("Message already processed successfully, messageId: {}", messageId);
                    return;
                }
                log.warn("Message is being consumed by another consumer, messageId: {}", messageId);
                return;
            }

            // 2. 提取消息内容
            if (body == null || body.isEmpty()) {
                log.error("Message body is null or empty, messageId: {}", messageId);
                messageQueueIdempotentHandler.delMessageProcessed(messageId); // 清除幂等标记以便重试
                return;
            }

            String blogIdStr = body.get("blogId");

            log.info("Extracted message fields - blogId: {}", blogIdStr);

            if (blogIdStr == null) {
                log.error("Invalid message format, missing required fields, messageId: {}", messageId);
                messageQueueIdempotentHandler.delMessageProcessed(messageId); // 清除幂等标记以便重试
                return;
            }

            // 3. 调用HTTP接口处理博客审核
            Long blogId;
            try {
                blogId = Long.parseLong(blogIdStr);
            } catch (NumberFormatException e) {
                log.error("Invalid blogId format: {}, messageId: {}", blogIdStr, messageId, e);
                messageQueueIdempotentHandler.delMessageProcessed(messageId); // 清除幂等标记以便重试
                return;
            }

            // 获取博客内容
            BlogDO blog = blogMapper.selectOne(new LambdaQueryWrapper<BlogDO>()
                    .eq(BlogDO::getId, blogId));
//            BlogDO blog = blogMapper.showBLogDetail(blogId);
            if (blog == null) {
                log.error("Blog not found with id: {}, messageId: {}", blogId, messageId);
                messageQueueIdempotentHandler.delMessageProcessed(messageId); // 清除幂等标记以便重试
                return;
            }

            log.info("Retrieved blog content for blog {}, title: {}", blogId, blog.getTitle());
            log.info("Calling review API for blog {}", blogId);
            ReviewResult result = callReviewApi(blog);

            if (result.isSuccess()) {
                // 4. 标记消息处理完成
                messageQueueIdempotentHandler.setAccomplish(messageId);
                log.info("Blog review processed successfully, blogId: {}", blogId);
            } else {
                // 审核不通过，添加审核不通过记录
                try {
                    AuditDO auditRecord = AuditDO.builder()
                            .action(AuditEnums.REJECTED.getValue()) // 1表示审核操作
                            .blogId(blogId)
                            .operatorId((long) AuditEnums.AUTO_REVIEW.getValue()) // 0表示系统操作
                            .reason("审核不通过: " + result.getMessage())
                            .createTime(new Date())
                            .updateTime(new Date())
                            .build();

                    auditMapper.insert(auditRecord);
                    log.info("Added audit rejection record for blogId: {}, reason: {}", blogId, result.getMessage());

                    // 标记消息处理完成（即使审核失败，也不重试）
                    messageQueueIdempotentHandler.setAccomplish(messageId);
                } catch (Exception auditException) {
                    log.error("Failed to insert audit rejection record for blogId: {}", blogId, auditException);
                    // 处理失败，删除幂等标识以便重试
                    messageQueueIdempotentHandler.delMessageProcessed(messageId);
                }
                log.error("Blog review rejected, blogId: {}, reason: {}", blogId, result.getMessage());
            }
        } catch (Exception e) {
            // 发生异常，删除幂等标识以便重试
            messageQueueIdempotentHandler.delMessageProcessed(messageId);
            log.error("Error processing blog review message: {}", messageId, e);
        }
    }

    /**
     * 调用博客审核API，支持重试机制
     *
     * @param blog 博客实体
     * @return 处理结果
     */
    @Retryable(value = {Exception.class}, maxAttempts = 3, backoff = @Backoff(delay = 1000, multiplier = 2))
    private ReviewResult callReviewApi(BlogDO blog) {
        long startTime = System.currentTimeMillis();
        int attempt = 1;
        try {
            log.info("Attempt #{} - Preparing to call review API for blogId: {}", attempt, blog.getId());

            // 1. 准备请求头
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);

            // 2. 准备请求体
            Map<String, Object> requestBody = new HashMap<>();
            requestBody.put("blogId", blog.getId());
            requestBody.put("title", blog.getTitle());
            requestBody.put("content", blog.getContent());
            requestBody.put("status", blog.getStatus());

            String requestBodyJson;
            try {
                requestBodyJson = objectMapper.writeValueAsString(requestBody);
                log.debug("Request body: {}", requestBodyJson);
            } catch (JsonProcessingException e) {
                log.error("Error serializing request body for blogId: {}", blog.getId(), e);
                return new ReviewResult(false, "序列化请求失败", null);
            }

            HttpEntity<String> entity = new HttpEntity<>(requestBodyJson, headers);

            // 3. 发送HTTP请求
            log.info("Sending HTTP request to {} for blogId: {}", reviewApiUrl, blog.getId());
            ResponseEntity<Map> response = restTemplate.postForEntity(reviewApiUrl, entity, Map.class);

            // 4. 处理响应
            long elapsed = System.currentTimeMillis() - startTime;
            if (response.getStatusCode().is2xxSuccessful()) {
                log.info("Review API call successful for blogId: {}, response: {}, took {}ms",
                        blog.getId(), response.getBody(), elapsed);
                return new ReviewResult(true, "审核请求成功处理", response.getBody());
            } else {
                log.error("Review API call failed for blogId: {}, status: {}, took {}ms",
                        blog.getId(), response.getStatusCode(), elapsed);
                return new ReviewResult(false, "审核请求失败: " + response.getStatusCode(), response.getBody());
            }
        } catch (Exception e) {
            long elapsed = System.currentTimeMillis() - startTime;
            log.error("Error calling review API for blogId: {}, took {}ms, exception: {}",
                    blog.getId(), elapsed, e.getMessage(), e);
            throw e; // 抛出异常以触发重试
        }
    }

    /**
     * 审核结果封装类
     */
    @Data
    @AllArgsConstructor
    public static class ReviewResult {
        private boolean success;
        private String message;
        private Map<String, Object> data;

        @Override
        public String toString() {
            return "ReviewResult{success=" + success + ", message='" + message + "', data=" + data + "}";
        }
    }
}
